package org.lds.message.center.aliyun;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.lds.message.center.aliyun.constant.AliyunSmsConstant;
import org.lds.message.center.aliyun.enums.AliyunSmsErrorEnum;
import org.lds.message.center.aliyun.properties.AliyunSmsProperties;
import org.lds.message.center.common.AbstractMessageService;
import org.lds.message.center.common.MessageHandler;
import org.lds.message.center.common.model.request.AliyunSmsRequest;
import org.lds.message.center.common.util.AssertUtil;
import org.lds.message.center.plugin.core.kafka.KafkaProducer;

import java.util.Objects;

@Slf4j
public class AliyunSmsMessageService extends AbstractMessageService<AliyunSmsRequest> {
    private final AliyunSmsProperties aliyunSmsProperties;
    private final KafkaProducer kafkaProducer;

    public AliyunSmsMessageService(MessageHandler<AliyunSmsRequest> messageHandler, AliyunSmsProperties aliyunSmsProperties, KafkaProducer kafkaProducer) {
        super(messageHandler);
        this.aliyunSmsProperties = aliyunSmsProperties;
        this.kafkaProducer = kafkaProducer;
    }

    @Override
    public String channelName() {
        return AliyunSmsConstant.CHANNEL_NAME;
    }

    @Override
    public String channelType() {
        return AliyunSmsConstant.CHANNEL_TYPE;
    }

    @Override
    public Class<AliyunSmsRequest> supportClass() {
        return AliyunSmsRequest.class;
    }

    @Override
    public void check(AliyunSmsRequest request) {
        AssertUtil.assertTrue(aliyunSmsProperties.isEnabled(), AliyunSmsErrorEnum.PLUGIN_CLOSED);
        AssertUtil.assertTrue(Objects.nonNull(request), AliyunSmsErrorEnum.REQUEST_INVALID);
        AssertUtil.assertTrue(StringUtils.isNotEmpty(request.getPhoneNumbers()), AliyunSmsErrorEnum.PHONE_NUMBERS_INVALID);
        AssertUtil.assertTrue(StringUtils.isNotEmpty(request.getSignName()), AliyunSmsErrorEnum.SIGN_NAME_INVALID);
        AssertUtil.assertTrue(StringUtils.isNotEmpty(request.getTemplateCode()), AliyunSmsErrorEnum.TEMPLATE_CODE_INVALID);
    }

    @Override
    protected void store(AliyunSmsRequest request) {
        kafkaProducer.send(aliyunSmsProperties.getTopic(), request);
    }

}
